package com.yanduo.tags

import ch.hsr.geohash.GeoHash
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.Row
import redis.clients.jedis.Jedis

/**
  *
  * @author Gerry chan
  * 2020/5/17 18:07
  * @version 1.0
  */
object Tags4Business extends Tags {
  override def makeTags(args: Any*): Map[String, Int] = {
    var map = Map[String, Int]()
    val row = args(0).asInstanceOf[Row]
    //获取jedis连接参数
    val jedis = args(1).asInstanceOf[Jedis]

    val lat = row.getAs[String]("lat")
    val longs = row.getAs[String]("long")

    //进行数据判断
    if (StringUtils.isNotEmpty(lat) && StringUtils.isNoneEmpty(longs)) {
      var lat2 = lat.toDouble
      var lon2 = longs.toDouble

      if(lat2 >3 && lat2<54 && lon2>73 && lon2<136) {
        val gethashCode = GeoHash.withCharacterPrecision(lat2, lon2, 8).toBase32
        val str = jedis.get(gethashCode)
        if (StringUtils.isNotEmpty(str)) {
          str.split(",").foreach(bs => map+="BS"+bs->1)
        }

      }
    }

    map
  }
}
